package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TestLevelBean;
import com.atguigu.bean.TestQuestionBean;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTradeTrademarkCategoryUserRefundWindow4 {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
                /*
                env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                env.setStateBackend(new HashMapStateBackend());
                env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                System.setProperty("HADOOP_USER_NAME", "atguigu");
                 */
        // TODO 3. 从 Kafka dwd_test_test_exam 主题读取考试主题的数据
        String topic = "dwd_test_test_exam1";
        String groupId = "dws_trade_trademark_category_user_refund_window4";
        DataStreamSource<String> dbStream = env.addSource(KafkaUtil.getKafkaConsumer(topic, groupId));

        // TODO 4. 转换数据结构
        SingleOutputStreamOperator<JSONObject> mapStream = dbStream.map(JSON::parseObject);
        SingleOutputStreamOperator<TestLevelBean> map = mapStream.map(new MapFunction<JSONObject, TestLevelBean>() {
            @Override
            public TestLevelBean map(JSONObject jsonObject) throws Exception {
                Double score = jsonObject.getDouble("score");
                Long ts = jsonObject.getLong("ts") * 1000;
                if (score <= 30) {
                    return TestLevelBean.builder()
                            .level("0-30")
                            .personCount(1L)
                            .ts(ts)
                            .build();
                } else if (score <= 60) {
                    return TestLevelBean.builder()
                            .level("30-60")
                            .personCount(1L)
                            .ts(ts)
                            .build();
                } else if (score <= 80) {
                    return TestLevelBean.builder()
                            .level("60-80")
                            .personCount(1L)
                            .ts(ts)
                            .build();
                } else {
                    return TestLevelBean.builder()
                            .level("80-100")
                            .personCount(1L)
                            .ts(ts)
                            .build();
                }

            }
        });


        // TODO 6. 设置水位线
        SingleOutputStreamOperator<TestLevelBean> withWaterMarkStream = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TestLevelBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<TestLevelBean>() {
                    @Override
                    public long extractTimestamp(TestLevelBean tradeTrademarkCategoryUserRefundBean, long l) {
                        return tradeTrademarkCategoryUserRefundBean.getTs();
                    }
                }));

        KeyedStream<TestLevelBean, String> keyedStream1 = withWaterMarkStream.keyBy(new KeySelector<TestLevelBean, String>() {
            @Override
            public String getKey(TestLevelBean testLevelBean) throws Exception {
                return testLevelBean.getLevel();
            }
        });




        // TODO 8. 开窗
        WindowedStream<TestLevelBean, String, TimeWindow> windowStream = keyedStream1.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // TODO 9. 聚合
        SingleOutputStreamOperator<TestLevelBean> reduceStream = windowStream.reduce(new ReduceFunction<TestLevelBean>() {
            @Override
            public TestLevelBean reduce(TestLevelBean testBean, TestLevelBean t1) throws Exception {
                testBean.setPersonCount(testBean.getPersonCount()+ t1.getPersonCount());
                return testBean;
            }
        }, new WindowFunction<TestLevelBean, TestLevelBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<TestLevelBean> iterable, Collector<TestLevelBean> collector) throws Exception {
                TestLevelBean tradeTrademarkCategoryUserRefundBean = iterable.iterator().next();
                tradeTrademarkCategoryUserRefundBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                tradeTrademarkCategoryUserRefundBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                tradeTrademarkCategoryUserRefundBean.setTs(System.currentTimeMillis());
                collector.collect(tradeTrademarkCategoryUserRefundBean);
            }
        });

        reduceStream.print("reduce>>>");


        env.execute(groupId);


    }
}
